package com.ds.infrastructure.hyperspace.container.endpoint.entry;

import com.ds.infrastructure.hyperspace.common.StateEnum;
import com.ds.infrastructure.hyperspace.common.TargetEndpointStreamStatus;
import com.ds.infrastructure.hyperspace.common.config.EntryNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.NettyEndpointNodeStream;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.http.NettyHttpHeaderNames;
import com.ds.infrastructure.hyperspace.container.internal.http.NettyHttpHeaderValues;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2Codec;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameTypeEnum;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameWrapper;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2GoAwayFrame;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.impl.HyperspaceTracerContext;
import com.ds.infrastructure.hyperspace.container.internal.util.*;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.HeadersUtils;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2HeadersFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import static com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum.HTTP2;
import static com.ds.infrastructure.hyperspace.common.StateEnum.CLIENT_CLOSED_CONNECTION;
import static com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameTypeEnum.*;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.HYPHEN;

/**
 * @author  Weishaoying
 */
@Slf4j
public class Http2EntryProtocolAdapter extends AbstractEntryProtocolAdapter {

    private static final int CLIENT_FIRST_STREAM_ID = 3;

    private static final int PREFACE_MESSAGE_SIZE = 3;

    private static final String HYPERSPACE_GRPC_PING_URI = "/GrpcPing/Ping";

    private final HyperspaceHttp2Codec hyperspaceHttp2Codec;

    private HyperspaceDuplexFlowContext connectionFlowContext;

    private HyperspaceDuplexFlowContext currentFlowContext;

    private List<Object> prefaceDataList = new ArrayList<>(PREFACE_MESSAGE_SIZE);

    private final IntObjectMap<HyperspaceDuplexFlowContext> duplexFlowContextMap = new IntObjectHashMap<>();

    private int dataFramePayloadSize = 0;

    public Http2EntryProtocolAdapter(NettyContainer nettyContainer,
                                     EntryNodeConfig entryNodeConfig,
                                     HyperspaceHttp2Codec hyperspaceHttp2Codec) {
        super(nettyContainer, entryNodeConfig);
        this.hyperspaceHttp2Codec = hyperspaceHttp2Codec;
    }

    @Override
    public void handleChannelRead(ChannelHandlerContext ctx, Object msg) {
        HyperspaceHttp2FrameWrapper hyperspaceHttp2FrameWrapper = (HyperspaceHttp2FrameWrapper) msg;
        HyperspaceHttp2FrameTypeEnum frameType = hyperspaceHttp2FrameWrapper.getFrameType();
        if (hyperspaceHttp2FrameWrapper.getStreamId() == 0) {
            if (WINDOW_UPDATE == frameType) {
                return;
            }
            if (GO_AWAY == frameType) {
                HyperspaceHttp2GoAwayFrame hyperspaceHttp2GoAwayFrame =
                        (HyperspaceHttp2GoAwayFrame) hyperspaceHttp2FrameWrapper.getFrameData();
                hyperspaceHttp2GoAwayFrame.getOriginGoAwayFrame().release();
                //测试：ClosedChannelException:null 问题解决方案-待验证
                ctx.channel().eventLoop().execute(()-> closeEntryChannel());
                return;
            }
            if (SETTINGS == frameType) {
                hyperspaceHttp2Codec.writeAndFlushSettingsAck(ctx);
            }
            if (SETTINGS_ACK == frameType) {
                hyperspaceHttp2Codec.writeAndFlushWindowUpdateFrame(
                        ctx, nettyContainer.getHttp2Config().getWindowUpdateInitIncrement());
                return;
            }
            if (PING == frameType || PING_ACK == frameType) {
                if (currentFlowContext != null && !currentFlowContext.isPending()) {
                    entryFlowOperation.handleInput(hyperspaceConnector, currentFlowContext, hyperspaceHttp2FrameWrapper);
                }
                return;
            }
            prefaceDataList.add(hyperspaceHttp2FrameWrapper);
            return;
        }
        currentFlowContext = requireDuplexFlowContext(hyperspaceHttp2FrameWrapper);
        if (currentFlowContext == null) {
            releaseData(hyperspaceHttp2FrameWrapper);
            return;
        }
        if (!isChannelOpen()) {
            if (msg == currentFlowContext.getRequestStartState()) {
                entryFlowOperation.interruptAndCloseContext(currentFlowContext, CLIENT_CLOSED_CONNECTION,
                        "Entry client has closed connection,client ip:" + clientIp);
            } else {
                releaseData(hyperspaceHttp2FrameWrapper);
            }
            return;
        }
        if (hyperspaceHttp2FrameWrapper.isEndOfStream()) {
            currentFlowContext.setInputComplete(true);
        }
        if (DATA == frameType) {
            dataFramePayloadSize += hyperspaceHttp2FrameWrapper.payloadSize();
            if (dataFramePayloadSize > nettyContainer.getHttp2Config().getWindowUpdateSendIncrement()) {
                hyperspaceHttp2Codec.writeAndFlushWindowUpdateFrame(ctx, dataFramePayloadSize);
                dataFramePayloadSize = 0;
            }
        }
        entryFlowOperation.handleInput(hyperspaceConnector, currentFlowContext, hyperspaceHttp2FrameWrapper);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
//        if (currentFlowContext != null) {
//            entryFlowOperation.handleInputComplete(hyperspaceConnector, currentFlowContext);
//        }
    }

    @Override
    public HyperspaceDuplexFlowContext currentDuplexFlowContext() {
        return currentFlowContext;
    }

    @Override
    public void resetChannelActiveState(ChannelHandlerContext ctx, HyperspaceDuplexFlowContext duplexFlowContext) {
        super.resetChannelActiveState(ctx, duplexFlowContext);
        this.connectionFlowContext = duplexFlowContext;
    }

    @Override
    public HyperspaceDuplexFlowContext requireDuplexFlowContext(int flowContextKey) {
        return duplexFlowContextMap.get(flowContextKey);
    }

    @Override
    public void writeToEntry(HyperspaceDuplexFlowContext duplexFlowContext, Object msg) {
        HyperspaceHttp2FrameWrapper frameWrapper = (HyperspaceHttp2FrameWrapper) msg;
        HyperspaceHttp2FrameTypeEnum frameType = frameWrapper.getFrameType();
        try {
            if (!isChannelOpen() || duplexFlowContext.isClosed()) {
                releaseData(frameWrapper);
                return;
            }
            if (SETTINGS_ACK == frameType || SETTINGS == frameType) {
                return;
            }
            if (PING == frameType || PING_ACK == frameType) {
                hyperspaceHttp2Codec.writeAndFlushData(channelHandlerContext, frameWrapper);
                return;
            }
            hyperspaceHttp2Codec.writeData(channelHandlerContext, frameWrapper);
        } finally {
            if (frameWrapper.isEndOfStream()) {
                if (HEADERS == frameType || Http2Util.isErrorCode(currentFlowContext.getTargetResponseCode())) {
                    duplexFlowContextMap.remove(frameWrapper.getStreamId());
                }
            }
        }
    }

    @Override
    protected void interruptToEntryComplete(HyperspaceDuplexFlowContext duplexFlowContext) {
        String tracerId = HYPHEN;
        if (duplexFlowContext.getTracerContext() != null) {
            tracerId = duplexFlowContext.getTracerContext().getTracerId();
        }
        StateEnum stateEnum = duplexFlowContext.getInterruptState();
        log.warn("Request was interrupted! state [{}],tracer id[{}], interrupt reason [{}], entry [{}:{}], host [{}], uri [{}]"
                , stateEnum.name(), tracerId, duplexFlowContext.getInterruptMessage(), entryNodeConfig.getHost()
                , entryNodeConfig.getPort(), duplexFlowContext.getHost(), duplexFlowContext.getUri());
        if (duplexFlowContext.getRequestStartState() instanceof HyperspaceHttp2FrameWrapper) {
            HyperspaceHttp2FrameWrapper frameWrapper = (HyperspaceHttp2FrameWrapper) duplexFlowContext.getRequestStartState();
            duplexFlowContextMap.remove(frameWrapper.getStreamId());
            if (isChannelOpen()) {
                String errorMessage = String.format(
                        "RT:request was interrupted!http code[%s],state[%s],tracerId:%s ,interrupt reason [%s]",
                        stateEnum.getHttpCode(), stateEnum.name(), tracerId, duplexFlowContext.getInterruptMessage());
                Http2Headers http2Header = new DefaultHttp2Headers();
                http2Header.add(NettyHttpHeaderNames.STATUS, Http2Util.newValue(String.valueOf(stateEnum.getHttpCode())));
                http2Header.add(NettyHttpHeaderNames.CONTENT_TYPE, NettyHttpHeaderValues.CONTENT_TYPE_GRPC);
                http2Header.add(NettyHttpHeaderNames.GRPC_STATUS, Http2Util.newValue(String.valueOf(stateEnum.getGrpcCode())));
                http2Header.add(NettyHttpHeaderNames.GRPC_MESSAGE, Http2Util.newValue(errorMessage));
                Http2HeadersFrame http2HeadersFrame = new DefaultHttp2HeadersFrame(http2Header, true);
                HyperspaceHttp2FrameWrapper grpcHttp2HeaderFrameWrapper =
                        new HyperspaceHttp2FrameWrapper(HEADERS, frameWrapper.getStreamId(), http2HeadersFrame);
                hyperspaceHttp2Codec.writeAndFlushData(channelHandlerContext, grpcHttp2HeaderFrameWrapper);
            }
        }
    }

    @Override
    public void releaseData(Object data) {
        if (data instanceof HyperspaceHttp2FrameWrapper) {
            HyperspaceHttp2FrameWrapper hyperspaceHttp2FrameWrapper = (HyperspaceHttp2FrameWrapper) data;
            NettyReleaseUtil.releaseData(hyperspaceHttp2FrameWrapper.getFrameData());
        }
    }

    @Override
    public void clearSuspendRequest() {
    }

    @Override
    public void releaseDuplexFlowContext() {
        Iterator<IntObjectMap.PrimitiveEntry<HyperspaceDuplexFlowContext>> duplexFlowContextIterator
                = duplexFlowContextMap.entries().iterator();
        while (duplexFlowContextIterator.hasNext()) {
            IntObjectMap.PrimitiveEntry<HyperspaceDuplexFlowContext> entry = duplexFlowContextIterator.next();
            HyperspaceDuplexFlowContext duplexFlowContext = entry.value();
            if (duplexFlowContext == null) {
                continue;
            }
            if (duplexFlowContext.isClosed()) {
                duplexFlowContextMap.remove(entry.key());
                continue;
            }
            if (!isChannelOpen()) {
                entryFlowOperation.interruptAndCloseContext(duplexFlowContext, CLIENT_CLOSED_CONNECTION,
                        "Entry client has closed connection,client ip[" + clientIp + "]");
                continue;
            }
            if (duplexFlowContext.getCurrentReferTargetEndpointStream() != null &&
                    duplexFlowContext.getCurrentReferTargetEndpointStream().getStatus() == TargetEndpointStreamStatus.CLOSED) {
                entryFlowOperation.interruptAndCloseContext(duplexFlowContext, CLIENT_CLOSED_CONNECTION,
                        "Target client has closed connection,target node info[" +
                                duplexFlowContext.getCurrentReferTargetEndpointStream().getTargetEndpointNode().getKey() + "]");
            }
        }
    }

    @Override
    public String getEntryHeader(HyperspaceDuplexFlowContext duplexFlowContext, String headerName) {
        if (duplexFlowContext == null) {
            return null;
        }
        if (headerName == null) {
            return null;
        }
        HyperspaceHttp2FrameWrapper http2HeadersFrameWrapper =
                (HyperspaceHttp2FrameWrapper) duplexFlowContext.getRequestStartState();
        Http2HeadersFrame http2HeadersFrame = (Http2HeadersFrame) http2HeadersFrameWrapper.getFrameData();
        return HeadersUtils.getAsString(http2HeadersFrame.headers(), headerName.toLowerCase());
    }

    @Override
    public byte[] getEntryBody(HyperspaceDuplexFlowContext duplexFlowContext) {
        return null;
    }

    private HyperspaceDuplexFlowContext requireDuplexFlowContext(HyperspaceHttp2FrameWrapper frameWrapper) {
        if (frameWrapper.getStreamId() < CLIENT_FIRST_STREAM_ID) {
            return connectionFlowContext;
        }
        if (HEADERS == frameWrapper.getFrameType()) {
            return initDuplexFlowContext(frameWrapper);
        }
        return duplexFlowContextMap.get(frameWrapper.getStreamId());
    }

    private HyperspaceDuplexFlowContext initDuplexFlowContext(HyperspaceHttp2FrameWrapper headerFrameWrapper) {
        Http2HeadersFrame http2HeadersFrame = (Http2HeadersFrame) headerFrameWrapper.getFrameData();
        Http2Headers http2Headers = http2HeadersFrame.headers();
        if (HYPERSPACE_GRPC_PING_URI.equals(http2Headers.path().toString())) {
            hyperspaceHttp2Codec.writeData(channelHandlerContext,
                    Http2Util.newSuccessHeadersFrameWrapper(headerFrameWrapper.getStreamId()));
            hyperspaceHttp2Codec.writeData(channelHandlerContext,
                    Http2Util.newDataFrameWrapper(headerFrameWrapper.getStreamId(), Http2Util.getGrpcPingResponseData()));
            hyperspaceHttp2Codec.writeAndFlushData(channelHandlerContext,
                    Http2Util.newEndHeadersFrameWrapper(headerFrameWrapper.getStreamId()));
            touchTargetEndpointStream();
            return null;
        }
        HyperspaceDuplexFlowContext duplexFlowContext = new HyperspaceDuplexFlowContext(
                nettyContainer, this, headerFrameWrapper,
                HeadersUtils.getAsString(http2Headers, NettyHttpHeaderNames.X_FORWARDED_FOR),
                HeadersUtils.getAsString(http2Headers, NettyHttpHeaderNames.X_REAL_IP),
                clientIp, HTTP2, connectionFlowContext);
        duplexFlowContext.setHost(http2Headers.authority().toString());
        duplexFlowContext.setUri(http2Headers.path().toString());
        duplexFlowContext.setRequestType(http2Headers.method().toString());
        duplexFlowContext.setEntryPrefaceState(prefaceDataList);
        String grpcService = Http2Util.getGrpcService(http2Headers.path().toString());
        if (StringUtils.isNotBlank(grpcService)) {
            duplexFlowContext.setGrpcServiceRouterName(Http2Util.getGrpcServiceRouterName(grpcService));
        }
        if (nettyContainer.getOpentracingConfig().isEnable()) {
            duplexFlowContext.setTracerContext(
                    new HyperspaceTracerContext(AddressesUtil.getIpAndPort(channel.localAddress()), clientIp, http2Headers));
            duplexFlowContext.getTracerContext()
                    .sample(duplexFlowContext.getRequestStartTime(), nettyContainer.getOpentracingConfig());
        }
        if (connectionFlowContext == null) {
            connectionFlowContext = duplexFlowContext;
        }
        duplexFlowContextMap.put(headerFrameWrapper.getStreamId(), duplexFlowContext);
        return duplexFlowContext;
    }

    private void touchTargetEndpointStream() {
        if (connectionFlowContext != null &&
                connectionFlowContext.getTargetEndpointStream() instanceof NettyEndpointNodeStream) {
            NettyEndpointNodeStream stream = (NettyEndpointNodeStream) connectionFlowContext.getTargetEndpointStream();
            stream.touch();
        }
    }

}
